package com.dxf.bigdata.D04_spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *  RDD :  弹性分布式数据集
 *   类比 T01中的Task
 *
 *                                       Task (理解为 T01_SubTask)                   Executor
 *   driver(RDD(可以理解为T01_Task))   ----------------------------------------->
 *                                                                                   Executor
 *
 *
 *   RDD 最小的计算单元,代表一个弹性,不可变,可分区,;面元素可并行的计算集合
 *
 *   弹性:
 *      1 存储弹性 :内存和磁盘自动切换
 *      2 容错弹性 :数据丢失自动回复
 *      3 计算弹性 :计算出错重试机制
 *      4 分片弹性 :可根据需求重新分片
 *   分布式:
 *      数据存储在大数据集群不同节点上
 *   数据集:
 *      RDD封装了计算逻辑,并不保存数据
 *   数据抽象:
 *      RDD是一个抽象类,需要子类具体实现
 *   不可变:
 *      RDD封装了计算逻辑,是不可改变的,想改变只能生产新的RDD
 *   可分区并行计算
 *
 *   =============================
 *   RDD 5个属性
 *  Internally, each RDD is characterized by five main properties:
 *  *
 *  *  - A list of partitions
 *  *  - A function for computing each split
 *  *  - A list of dependencies on other RDDs
 *  *  - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
 *  *  - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
 *  *    an HDFS file)
 *
 *   1 分区列表
 *   2 分区计算函数
 *   3 RDD依赖关系 链表
 *   4 分区器  数据如何分区
 *   5 首选位置   数据发送给哪一个Executor--> 发送给哪一个节点进行计算效率最优(如 移动数据不如移动计算)   getPreferredLocations
 *
 *   =====================================================
 *
 *
 *   RDD 创建:
 *     1 从内存中创建RDD
 *
 *     2 文件中创建RDD
 *
 *     =========================================================
 *
 *     RDD的并行度和分区:
 *      1 可以通过设置确认 并行度和分区  ->先有分区, 每个分区中可以有一个 task ,
 *
 *
 *
 *
 */
object T06_RDD并行度和分区_分区后数据的分配 {
  def main(args: Array[String]): Unit = {

    //TODO 准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    //TODO 创建RDD
    // minPartitions: Int = defaultMinPartitions 最小分区,可以指定
    // 1 数据以行为单位进行读取
    // spark采用的方式和handoop一样,一行一行读取
    // 2 数据读取时以偏移量为单位
    /**
     * 偏移量理解:
     *  1@@    =>012
     *  2@@    =>345
     *  3      =>6
     *
     */
      //TODO  3  偏移量计算 有7个字节 ,最小分区2 ,则每个分区个数  7/2 =3
      //偏移量不会被重复读取
      //第一分区 0 => [0 , 3]
      //第二分区 1 => [3 , 6]
      //第三分区 2 => [6 , 7]
    val rdd: RDD[String] = sc.textFile("datas/1.txt",2)

    rdd.saveAsTextFile("output")
    //TODO 关闭环境
    sc.stop()


  }
}
